package com.west.xu.tdmq.controller;

import com.west.xu.tdmq.domain.Notice;
import io.swagger.annotations.ApiImplicitParam;
import io.swagger.annotations.ApiImplicitParams;
import io.swagger.annotations.ApiOperation;
import org.apache.commons.lang.RandomStringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;

import java.util.concurrent.TimeUnit;

@RestController
public class SendMsgController {

    private static final Log logger = LogFactory.getLog(SendMsgController.class);


    @Autowired
    private Producer<Notice> noticeProducer;


    @RequestMapping(value = "/send/{msg}", method = RequestMethod.GET)
    @ApiOperation(value = "发送信息", notes = "发送信息到TDMQ")
    @ApiImplicitParams({
            @ApiImplicitParam(paramType = "path", name = "msg", value = "待发送信息", dataType = "String", required = true)
    })
    public String send(@PathVariable String msg) {
        MessageId messageId = null;
        Notice notice = new Notice(RandomStringUtils.randomNumeric(6), msg);
        try {
            messageId = noticeProducer.send(notice);
            logger.info("发送消息Success===>" + notice.toString() + "|MessageId===>" + messageId.toString());
        } catch (PulsarClientException e) {
            logger.info("发送消息Failure===>" + notice.toString() + "|Ex===>" + e.getMessage());
        }
        return "Success";
    }

    @RequestMapping(value = "/send/{key}/{msg}", method = RequestMethod.GET)
    @ApiOperation(value = "发送信息", notes = "发送信息到TDMQ,指定Key值")
    @ApiImplicitParams({
            @ApiImplicitParam(paramType = "path", name = "key", value = "待发送信息Key", dataType = "String", required = true),
            @ApiImplicitParam(paramType = "path", name = "msg", value = "待发送信息", dataType = "String", required = true)
    })
    public String send(@PathVariable String key, @PathVariable String msg) {
        MessageId messageId = null;
        Notice notice = new Notice(RandomStringUtils.randomNumeric(6), msg);
        try {
            messageId = noticeProducer
                    .newMessage(Schema.JSON(Notice.class))
                    .key(key)
                    .value(notice)
                    .send();
            logger.info("发送消息Success===>key:" + key + " |  msg:" + notice.toString() + "|MessageId===>" + messageId.toString());
        } catch (PulsarClientException e) {
            logger.info("发送消息Failure===>key:" + key + " |  msg:" + notice.toString() + "|Ex===>" + e.getMessage());
        }
        return "Success";
    }

    @RequestMapping(value = "/delay/{msg}/{delay}", method = RequestMethod.GET)
    @ApiOperation(value = "发送信息", notes = "发送信息到TDMQ,延迟N分钟")
    @ApiImplicitParams({
            @ApiImplicitParam(paramType = "path", name = "msg", value = "待发送信息", dataType = "String", required = true),
            @ApiImplicitParam(paramType = "path", name = "delay", value = "延迟时间", dataType = "Long", required = true)
    })
    public String sendDelay(@PathVariable String msg, @PathVariable Long delay) {
        MessageId messageId = null;
        Notice notice = new Notice(RandomStringUtils.randomNumeric(6), msg);
        try {
            messageId = noticeProducer
                    .newMessage(Schema.JSON(Notice.class))
                    .deliverAfter(delay, TimeUnit.MINUTES)
                    .value(notice)
                    .send();
            logger.info("延迟发送消息Success===>Delay:" + delay + "Minutes |  msg:" + notice.toString() + "|MessageId===>" + messageId.toString());
        } catch (PulsarClientException e) {
            logger.info("延迟发送消息Failure===>Delay:" + delay + "Minutes |  msg:" + notice.toString() + "|Ex===>" + e.getMessage());
        }
        return "Success";
    }

    @RequestMapping(value = "/notice", method = RequestMethod.POST)
    @ApiOperation(value = "发送信息", notes = "发送信息到TDMQ")
    public String notice(@RequestBody Notice notice) {
        MessageId messageId = null;
        try {
            messageId = noticeProducer.send(notice);
            logger.info("发送消息Success===>" + notice.toString() + "|MessageId===>" + messageId.toString());
        } catch (PulsarClientException e) {
            logger.info("发送消息Failure===>" + notice.toString() + "|Ex===>" + e.getMessage());
        }
        return "Success";
    }
}
